package com.och.mrcp.server;

import com.och.mrcp.config.NettyConfig;
import com.och.mrcp.core.rtp.session.RtpSessionManager;
import com.och.mrcp.handler.RtpPacketHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import io.netty.channel.FixedRecvByteBufAllocator;

/**
 * RTP服务器
 * 支持动态端口绑定，通过SIP协议协商确定端口
 * @author danmo
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RtpServer {
    
    private final NettyConfig nettyConfig;
    private final RtpSessionManager sessionManager;
    private final RtpPacketHandler rtpPacketHandler;

    private EventLoopGroup workerGroup;
    private final Map<Integer, Channel> portChannels = new ConcurrentHashMap<>();
    private final AtomicBoolean isRunning = new AtomicBoolean(false);
    
    // 使用读写锁替代synchronized，提高并发性能
    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();
    
    /**
     * 启动RTP服务器
     * 初始化工作线程组，不绑定端口
     * RTP端口通过SIP协议协商动态分配
     */
    @PostConstruct
    public void start() {
        writeLock.lock();
        try {
            if (isRunning.get()) {
                log.warn("RTP server is already running");
                return;
            }
            
            try {
                log.info("Starting RTP server (initializing worker threads only)...");
                
                // 只初始化工作线程组，不绑定端口
                // RTP端口将通过SIP协议协商后动态分配
                workerGroup = new NioEventLoopGroup(nettyConfig.getWorkerThreads());
                isRunning.set(true);
                
                log.info("RTP server initialized successfully (no ports bound yet)");
                
            } catch (Exception e) {
                log.error("Failed to start RTP server: {}", e.getMessage(), e);
                cleanup();
                throw new RuntimeException("RTP server startup failed", e);
            }
        } finally {
            writeLock.unlock();
        }
    }
    
    /**
     * 在指定端口启动RTP监听
     * 通过SIP协议协商确定端口
     */
    public boolean startPort(int port) {
        readLock.lock();
        try {
            if (!isRunning.get()) {
                log.warn("RTP server not running, start it first");
                return false;
            }
            
            if (portChannels.containsKey(port)) {
                log.warn("RTP server already listening on port {}", port);
                return true;
            }
            
            try {
                log.info("Starting RTP listener on port {}", port);
                
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(workerGroup)
                        .channel(NioDatagramChannel.class)
                        .option(ChannelOption.SO_BROADCAST, true)
                        .option(ChannelOption.SO_RCVBUF, 131072) // 128KB接收缓冲区（优化）
                        .option(ChannelOption.SO_SNDBUF, 131072) // 128KB发送缓冲区（优化）
                        .option(ChannelOption.SO_REUSEADDR, true) // 端口复用
                        .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(2048)) // 固定缓冲区分配器
                        .handler(rtpPacketHandler);
                
                // 异步绑定，优先绑定到 IPv4 地址，避免阻塞主线程
                InetSocketAddress bindAddress = new InetSocketAddress(port);
                bootstrap.bind(bindAddress).addListener(future -> {
                    if (future.isSuccess()) {
                        Channel channel = ((io.netty.channel.ChannelFuture) future).channel();
                        portChannels.put(port, channel);
                        log.info("RTP listener started successfully on port {} ({}), handler: {}", 
                                port, channel.localAddress(), rtpPacketHandler.getClass().getSimpleName());
                    } else {
                        log.error("Failed to bind RTP listener on port {}: {}", 
                                port, future.cause().getMessage());
                    }
                });
                
                log.info("RTP listener started successfully on port {}", port);
                return true;
                
            } catch (Exception e) {
                log.error("Failed to start RTP listener on port {}: {}", port, e.getMessage(), e);
                return false;
            }
        } finally {
            readLock.unlock();
        }
    }
    
    /**
     * 停止指定端口的RTP监听
     */
    public boolean stopPort(int port) {
        readLock.lock();
        try {
            Channel channel = portChannels.get(port);
            if (channel == null) {
                log.warn("RTP server not listening on port {}", port);
                return false;
            }
            
            try {
                log.info("Stopping RTP listener on port {}", port);
                
                // 异步关闭，避免阻塞
                channel.close().addListener(future -> {
                    if (future.isSuccess()) {
                        log.info("RTP listener on port {} closed successfully", port);
                    } else {
                        log.warn("Error closing RTP listener on port {}: {}", port, future.cause().getMessage());
                    }
                });
                
                portChannels.remove(port);
                log.info("RTP listener stopped successfully on port {}", port);
                return true;
                
            } catch (Exception e) {
                log.error("Error stopping RTP listener on port {}: {}", port, e.getMessage(), e);
                return false;
            }
        } finally {
            readLock.unlock();
        }
    }
    
    /**
     * 停止RTP服务器
     */
    public void stop() {
        writeLock.lock();
        try {
            if (!isRunning.get()) {
                log.warn("RTP server is not running");
                return;
            }
            
            try {
                log.info("Stopping RTP server...");
                
                // 停止所有端口的监听（异步关闭，避免阻塞）
                for (Map.Entry<Integer, Channel> entry : portChannels.entrySet()) {
                    try {
                        entry.getValue().close().addListener(future -> {
                            if (future.isSuccess()) {
                                log.debug("Channel on port {} closed successfully", entry.getKey());
                            } else {
                                log.warn("Error closing channel on port {}: {}", entry.getKey(), future.cause().getMessage());
                            }
                        });
                    } catch (Exception e) {
                        log.warn("Error closing channel on port {}: {}", entry.getKey(), e.getMessage());
                    }
                }
                portChannels.clear();
                
                if (workerGroup != null) {
                    // 异步关闭工作线程组，避免阻塞
                    workerGroup.shutdownGracefully().addListener(future -> {
                        if (future.isSuccess()) {
                            log.debug("Worker group shut down successfully");
                        } else {
                            log.warn("Error shutting down worker group: {}", future.cause().getMessage());
                        }
                    });
                    workerGroup = null;
                }
                
                // 关闭会话管理器
                sessionManager.shutdown();
                
                isRunning.set(false);
                log.info("RTP server stopped successfully");
                
            } catch (Exception e) {
                log.error("Error stopping RTP server: {}", e.getMessage(), e);
            } finally {
                cleanup();
            }
        } finally {
            writeLock.unlock();
        }
    }
    
    /**
     * 清理资源
     */
    private void cleanup() {
        // 清理所有端口通道
        for (Map.Entry<Integer, Channel> entry : portChannels.entrySet()) {
            try {
                entry.getValue().close();
            } catch (Exception e) {
                log.warn("Error closing channel on port {}: {}", entry.getKey(), e.getMessage());
            }
        }
        portChannels.clear();
        
        if (workerGroup != null) {
            try {
                workerGroup.shutdownGracefully();
            } catch (Exception e) {
                log.warn("Error shutting down worker group: {}", e.getMessage());
            }
            workerGroup = null;
        }
        
        isRunning.set(false);
    }
    
    /**
     * 检查服务器是否正在运行
     */
    public boolean isRunning() {
        return isRunning.get();
    }
    
    /**
     * 检查指定端口是否正在监听
     */
    public boolean isPortListening(int port) {
        return portChannels.containsKey(port);
    }
    
    /**
     * 获取所有监听的端口
     */
    public Set<Integer> getListeningPorts() {
        return new HashSet<>(portChannels.keySet());
    }
    
    /**
     * 获取会话管理器
     */
    public RtpSessionManager getSessionManager() {
        return sessionManager;
    }
    
    /**
     * 检查是否有活跃的RTP会话
     */
    public boolean hasActiveSessions() {
        return sessionManager.getActiveSessionCount() > 0;
    }
    
    /**
     * 优雅关闭
     * 在应用关闭时调用
     */
    public void shutdown() {
        stop();
    }
}

